package belf.migrate.engine.job.sql;

import belf.migrate.api.job.Job;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.engine.BelfEvn;
import belf.migrate.engine.job.*;
import belf.migrate.api.taskconf.Task;
import lombok.extern.slf4j.Slf4j;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class SyncSpeicalRowsRunner extends BaseJobRunner {

    public SyncSpeicalRowsRunner(Job job) {
        super(job);
    }

    @Override
    public JobResult check() {
        JobResult jobResult = new JobResult(SUCCESS, "任务执行成功");

        jobContext = new JobContext("" + job.getJobId(), taskConf);
        // 创建源端catalog
        AbstractJdbcCatalog SourceCatalog = CatalogCreator.createSourceCatalog(taskConf);
        if (null == SourceCatalog) {
            log.error("No Source Catalog for {}, Stop Job {}!",
                    taskConf.getSourceConf().getCatalogName(),
                    job.getJobId());

            jobResult = new JobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到目标端" +
                    taskConf.getSinkConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");

            return jobResult;
        }
        jobContext.setSourceCatalog(SourceCatalog);
        BelfEvn.getInstance().addJobContext("" + job.getJobId(), jobContext);

        return jobResult;
    }

    @Override
    public JobResult execute() {
        JobResult jobResult = check();
        if (jobResult.getCode() != SUCCESS) {
            return jobResult;
        }

        if (job.getTask().getTaskType() != Task.TaskType.SYNC_SPECIAL_ROWS) {
            return new JobResult(FAILTURE, "任务类型不正确");
        }

        QualifiersModel qualifiersModel = jobContext.getSourceQualifiersModel();
        /*
        try {
            AbstractJdbcCatalog sourceCatalog = CatalogCreator.createSourceCatalog(taskConf);
            TablePath tablePath = TablePath.of(null, "OE", "CUSTOMERS", qualifiersModel);
            String[] columns = {"customer_id", "cust_address"};
            ResultSet rs = sourceCatalog.querySomeColumns(tablePath, columns);
            while (rs.next()) {
                System.out.println(rs.getLong("customer_id"));
                Object address = rs.getObject("cust_address");
                Binding binding = new Binding();
                binding.setProperty("address", address);
                GroovyShell shell = new GroovyShell(binding);
                Object result = shell.evaluate("def struct = address.getAttributes()\n println \"$struct\"");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        */
        return jobResult;
    }
}
